package com.aoip.telemetry.telemetry.receive.handler;

import com.aoip.telemetry.telemetry.core.redis.RedisCache;
import com.aoip.telemetry.telemetry.entity.constant.TelemetryConstants;
import com.aoip.telemetry.telemetry.entity.enums.AbnoTypeEnum;
import com.aoip.telemetry.telemetry.entity.enums.CacheKeyEnum;
import com.aoip.telemetry.telemetry.entity.equipment.StOnlineStatusEntity;
import com.aoip.telemetry.telemetry.mq.rabbitmq.SendMsgToRabbitMq;
import com.aoip.telemetry.telemetry.receive.call.controller.QuerySetOrderController;
import com.aoip.telemetry.telemetry.receive.session.Session;
import com.aoip.telemetry.telemetry.receive.session.SessionFactory;
import com.aoip.telemetry.telemetry.receive.util.ConfirmUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicStampedReference;

/**
 * @author chenbai
 * @version 1.0
 * @date 2022/2/23 21:03
 */
@Component
@Slf4j
public class UdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    private static UdpServerHandler udpServerHandler;

    private static DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    @Resource
    private RedisCache redisCache;
    @Resource
    private SendMsgToRabbitMq sendMsgToMq;
    @Resource
    private QuerySetOrderController querySetOrderController;
    @Value("${receive.msg.max-length}")
    private int msgMaxLength;
    @PostConstruct
    public void init() {
        udpServerHandler = this;
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) {
        ByteBuf buf = packet.copy().content();
        byte[] req = new byte[buf.readableBytes()];
        if(req.length > udpServerHandler.msgMaxLength) {
            return;
        }
        buf.readBytes(req);
        String originalMessage = new String(req);
        if (!StringUtils.hasText(originalMessage)) {
            return;
        }
        log.error("UDP接收报文：{}", originalMessage);
        InetSocketAddress inetSocketAddress = packet.sender();
        test(originalMessage, ctx, inetSocketAddress);
    }

    private void test(String msg, ChannelHandlerContext ctx, InetSocketAddress inetSocketAddress) {
        String stCode = getStCode(msg);
        if (!StringUtils.hasText(stCode)) {
            return;
        }
        if (unregisteredStCode(stCode)) {
            return;
        }
        int port = inetSocketAddress.getPort();
        if (msg.contains("PIC")) {
            blackCheck(msg, ctx);
            return;
        }
        Session session = SessionFactory.getSession();
        Channel channel = ctx.channel();
        if (channel == null) {
            log.error("stCode={},{}", stCode, "channel");
            return;
        }
        session.bind(channel, stCode);
//        session.unbind(channel);
        //        保存测站在线信息到redis
        StOnlineStatusEntity entity = new StOnlineStatusEntity();
        entity.setStCode(stCode);
        entity.setOnlineStatus(TelemetryConstants.ONLINE);
        entity.setOnlineTime(LocalDateTime.now());
        entity.setCommunicateIpPort(port);
        String key = CacheKeyEnum.STATION_ONLINE_STATUS + ":" + stCode;
        udpServerHandler.redisCache.deleteObject(key);
        udpServerHandler.redisCache.setCacheObject(key, entity, 10, TimeUnit.MINUTES);

        udpServerHandler.redisCache.deleteObject(CacheKeyEnum.ST_GP_ABNO_INFO + ":" + AbnoTypeEnum.LAST_HEARTBEAT + ":" + stCode);
        udpServerHandler.redisCache.setCacheObject(CacheKeyEnum.ST_GP_ABNO_INFO + ":" + AbnoTypeEnum.LAST_HEARTBEAT + ":" + stCode, LocalDateTime.now().format(dtf));

        session.setAttribute(channel, "stCode", stCode);
        session.setAttribute(channel, "protocolCommunicate", TelemetryConstants.UDP);
        session.setAttribute(channel, "inSocket", inetSocketAddress);

        String confirm = ConfirmUtil.generateConfirm(msg);
        Collection<String> keys = udpServerHandler.redisCache.keys(CacheKeyEnum.ORDER_NO_SEND_STORE_HOUSE + ":" + stCode + ":*");
        if (StringUtils.hasText(confirm)) {
            ByteBuf heapBuffer = channel.alloc().heapBuffer();
            heapBuffer.writeBytes(confirm.getBytes());
            DatagramPacket sendPacket = new DatagramPacket(heapBuffer, inetSocketAddress);
            if (msg.length() != 18) {
                ctx.writeAndFlush(sendPacket);
            } else {
                if (keys == null || keys.size() < 1) {
                    ctx.writeAndFlush(sendPacket);
                }
            }
        }
        if (msg.length() == 18) {
            if (keys != null && keys.size() > 0) {
                //            心跳报文自动发送指令
                udpServerHandler.querySetOrderController.autoSend(stCode);
            }
        } else {
            udpServerHandler.sendMsgToMq.sendMsg("exchange_telemetry_msg", "router_key_telemetry_hydrology", msg);
        }
    }

    private String getStCode(String originalMessage) {
        int st = originalMessage.indexOf("ST");
        int stCodeBeginIndex = st + 3;
        int stCodeEndIndex = originalMessage.indexOf(" ", 3);
        return originalMessage.substring(stCodeBeginIndex, stCodeEndIndex);
    }

    /**
     * 未注册的测站，不存在，返回true，存在返回false
     *
     * @param stCode 测站编码
     * @return java.lang.Boolean
     * @author chenbai
     * @date 2022/11/25
     **/
    private boolean unregisteredStCode(String stCode) {
        Collection<String> keys = udpServerHandler.redisCache.keys(CacheKeyEnum.STATION + ":*" + stCode);
        return keys == null || keys.size() < 1;
    }

    private void blackCheck(String msg, ChannelHandlerContext ctx){
        String stCode = getStCode(msg);
        SocketAddress socketAddress = ctx.channel().localAddress();
        int serverPort = 0;
        if (socketAddress instanceof InetSocketAddress) {
            InetSocketAddress address = (InetSocketAddress)socketAddress;
            serverPort = address.getPort();
        }
//            String hostString = inetSocketAddress.getHostString();
//            String key = CacheKeyEnum.CLIENT_IP_PORT + ":" + hostString;
//            Set<String> blackIp = udpServerHandler.redisCache.getCacheObject(key);
//            if (blackIp == null || blackIp.size() < 1) {
//                blackIp = new HashSet<>();
//            }
//            blackIp.add(stCode);
        String blackStCodeKey = CacheKeyEnum.BLACK_ST_CODE.getCode() + ":" + stCode;
        Integer blackStCodeNum = udpServerHandler.redisCache.getCacheObject(blackStCodeKey);
        if (blackStCodeNum == null) {
            blackStCodeNum = 0;
        }
        udpServerHandler.redisCache.setCacheObject(blackStCodeKey, ++blackStCodeNum);
        String blackKey = CacheKeyEnum.BLACK_ST_CODE_LIST.getCode() + ":" + serverPort;
        Object blackIpObject = udpServerHandler.redisCache.getCacheObject(blackKey);
        Set<String> blackIpSet = null;
        if (blackIpObject == null) {
            blackIpSet = new HashSet<>();
        } else {
            if (blackIpObject instanceof Set) {
                blackIpSet = (Set)blackIpObject;
            }
        }
        if (blackIpSet == null || blackIpSet.size() < 1) {
            blackIpSet = new HashSet<>();
        }
        blackIpSet.add(stCode);
        udpServerHandler.redisCache.setCacheObject(blackKey, blackIpSet, 1, TimeUnit.DAYS);
    }
}
